package grpc

import (
	"context"
	"fmt"
	"time"

	"gitee.com/zawei/yggdrasil/pkg"
	"gitee.com/zawei/yggdrasil/pkg/errors"
	"gitee.com/zawei/yggdrasil/pkg/log"
	ymeta "gitee.com/zawei/yggdrasil/pkg/metadata"
	"gitee.com/zawei/yggdrasil/pkg/metrics"
	"gitee.com/zawei/yggdrasil/pkg/trace"
	"gitee.com/zawei/yggdrasil/pkg/transport/server"
	"github.com/opentracing/opentracing-go"
	"github.com/opentracing/opentracing-go/ext"
	traceLog "github.com/opentracing/opentracing-go/log"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/peer"
)

var (
	errSlowCommand = fmt.Errorf("grpc unary slow command")
)

// metric统计
func metricUnaryClientInterceptor(name string) func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		beg := time.Now()
		err := invoker(ctx, method, req, reply, cc, opts...)

		code := errors.FromError(err).HttpCode()
		codeStr := fmt.Sprintf("%d", code)
		_ = metrics.CounterAdd(server.MetricsHandleTotal, 1, map[string]string{
			"type": "grpc_unary", "name": name, "method": method, "peer": cc.Target(), "code": codeStr,
		})
		_ = metrics.HistogramObserve(server.MetricsHandleSec, time.Since(beg).Seconds(), map[string]string{
			"type": "grpc_unary", "name": name, "method": method, "peer": cc.Target(),
		})
		return err
	}
}

func metricStreamClientInterceptor(name string) func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
		beg := time.Now()
		clientStream, err := streamer(ctx, desc, cc, method, opts...)

		code := errors.FromError(err).HttpCode()
		codeStr := fmt.Sprintf("%d", code)
		_ = metrics.CounterAdd(server.MetricsHandleTotal, 1, map[string]string{
			"type": "grpc_unary", "name": name, "method": method, "peer": cc.Target(), "code": codeStr,
		})
		_ = metrics.HistogramObserve(server.MetricsHandleSec, time.Since(beg).Seconds(), map[string]string{
			"type": "grpc_unary", "name": name, "method": method, "peer": cc.Target(),
		})
		return clientStream, err
	}
}

func injectSpanContext(ctx context.Context, md metadata.MD) context.Context {
	clientSpan := opentracing.SpanFromContext(ctx)
	mdWriter := trace.TextArrayMapCarrier(md)
	err := opentracing.GlobalTracer().Inject(clientSpan.Context(), opentracing.HTTPHeaders, mdWriter)
	// We have no better place to record an error than the Span itself :-/
	if err != nil {
		clientSpan.LogFields(traceLog.String("event", "Tracer.Inject() failed"), traceLog.Error(err))
	}
	return metadata.NewOutgoingContext(ctx, md)
}

func traceUnaryClientInterceptor() grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		md, ok := metadata.FromOutgoingContext(ctx)
		if !ok {
			md = metadata.New(nil)
		} else {
			md = md.Copy()
		}
		span, ctx := opentracing.StartSpanFromContext(ctx, method, ext.SpanKindRPCClient)
		ext.Component.Set(span, "grpc")
		defer span.Finish()
		err := invoker(injectSpanContext(ctx, md), method, req, reply, cc, opts...)
		if err != nil {
			code := errors.FromError(err).HttpCode()
			span.SetTag("response_code", code)
			ext.Error.Set(span, true)
			span.LogFields(traceLog.String("event", "error"), traceLog.String("message", err.Error()))
		}
		return err
	}
}

func aidUnaryClientInterceptor() grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		md, ok := metadata.FromOutgoingContext(ctx)
		clientAidMD := metadata.Pairs("aid", pkg.AppID())
		if ok {
			md = metadata.Join(md, clientAidMD)
		} else {
			md = clientAidMD
		}
		ctx = metadata.NewOutgoingContext(ctx, md)

		return invoker(ctx, method, req, reply, cc, opts...)
	}
}

// timeoutUnaryClientInterceptor gRPC客户端超时拦截器
func timeoutUnaryClientInterceptor(timeout time.Duration, slowThreshold time.Duration) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		now := time.Now()
		// 若无自定义超时设置，默认设置超时
		_, ok := ctx.Deadline()
		if !ok {
			var cancel context.CancelFunc
			ctx, cancel = context.WithTimeout(ctx, timeout)
			defer cancel()
		}

		err := invoker(ctx, method, req, reply, cc, opts...)
		du := time.Since(now)
		remoteIP := "unknown"
		if remote, ok := peer.FromContext(ctx); ok && remote.Addr != nil {
			remoteIP = remote.Addr.String()
		}

		if slowThreshold > time.Duration(0) && du > slowThreshold {
			log.Warnw(errSlowCommand.Error(), "method", method, "name", cc.Target(), "cost", du, "remote", remoteIP)
		}
		return err
	}
}

// loggerUnaryClientInterceptor gRPC客户端日志中间件
func loggerUnaryClientInterceptor(name string, accessInterceptorLevel string) grpc.UnaryClientInterceptor {
	return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
		beg := time.Now()
		err := invoker(ctx, method, req, reply, cc, opts...)
		if err != nil {
			st := errors.FromProto(errors.Convert(err).Proto())
			code := st.HttpCode()
			if st.HttpCode() >= 500 {
				// 只记录系统级别错误
				log.Errorw("access", "type", "unary", "code", code, "msg", st.Message(), "name", name,
					"method", method, "cost", time.Since(beg), "req", req, "res", reply)
			} else {
				// 业务报错只做warning
				log.Warnw("access", "type", "unary", "code", code, "msg", st.Message(), "name", name,
					"method", method, "cost", time.Since(beg), "req", req, "res", reply)
			}
			return err
		} else {
			if accessInterceptorLevel == "info" {
				log.Infow("access", "type", "unary", "code", 200, "name", name, "method", method,
					"cost", time.Since(beg), "req", req, "res", reply)
			}
		}

		return nil
	}
}

func errorsUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{},
	cc *grpc.ClientConn, invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption,
) error {
	err := invoker(ctx, method, req, reply, cc, opts...)
	if err != nil {
		st := errors.FromProto(errors.Convert(err).Proto())
		return st.Err()
	}
	return err
}

func errorsStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc,
	cc *grpc.ClientConn, method string, streamer grpc.Streamer,
	opts ...grpc.CallOption,
) (grpc.ClientStream, error) {
	cs, err := streamer(ctx, desc, cc, method, opts...)
	if err != nil {
		st := errors.FromProto(errors.Convert(err).Proto())
		return cs, st.Err()
	}
	return cs, err
}

func mdUnaryClientInterceptor(ctx context.Context, method string, req, reply interface{},
	cc *grpc.ClientConn, invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption,
) error {
	meta, ok := ymeta.FromOutContext(ctx)
	if ok {
		if m, ok := metadata.FromOutgoingContext(ctx); ok {
			ctx = metadata.NewOutgoingContext(ctx, metadata.Join(m, metadata.MD(meta)))
		} else {
			ctx = metadata.NewOutgoingContext(ctx, metadata.MD(meta))
		}
	}

	return invoker(ctx, method, req, reply, cc, opts...)
}

func mdStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc,
	cc *grpc.ClientConn, method string, streamer grpc.Streamer,
	opts ...grpc.CallOption,
) (grpc.ClientStream, error) {
	meta, ok := ymeta.FromOutContext(ctx)
	if ok {
		if m, ok := metadata.FromOutgoingContext(ctx); ok {
			ctx = metadata.NewOutgoingContext(ctx, metadata.Join(m, metadata.MD(meta)))
		} else {
			ctx = metadata.NewOutgoingContext(ctx, metadata.MD(meta))
		}
	}

	return streamer(ctx, desc, cc, method, opts...)
}
